package runtime

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"sync"
	"time"

	"github.com/asynkron/protoactor-go/actor"
	"github.com/asynkron/protoactor-go/remote"
	"github.com/google/uuid"
	"github.com/louloulin/dataflare/pkg/cluster"
	"github.com/louloulin/dataflare/pkg/cluster/coordination"
	"github.com/louloulin/dataflare/pkg/cluster/resource"
)

// NodeRole 表示节点角色
type NodeRole string

const (
	// NodeRoleManager 表示管理节点角色
	NodeRoleManager NodeRole = "manager"
	// NodeRoleWorker 表示工作节点角色
	NodeRoleWorker NodeRole = "worker"
	// NodeRoleMixed 表示混合节点角色
	NodeRoleMixed NodeRole = "mixed"
)

// NodeStatus 表示节点状态
type NodeStatus string

const (
	// NodeStatusInitializing 表示节点正在初始化
	NodeStatusInitializing NodeStatus = "initializing"
	// NodeStatusReady 表示节点已就绪
	NodeStatusReady NodeStatus = "ready"
	// NodeStatusBusy 表示节点繁忙
	NodeStatusBusy NodeStatus = "busy"
	// NodeStatusDraining 表示节点正在排空
	NodeStatusDraining NodeStatus = "draining"
	// NodeStatusMaintenance 表示节点处于维护模式
	NodeStatusMaintenance NodeStatus = "maintenance"
	// NodeStatusError 表示节点出错
	NodeStatusError NodeStatus = "error"
)

// NodeConfig 表示节点配置
type NodeConfig struct {
	// ID 是节点ID
	ID string
	// Name 是节点名称
	Name string
	// Role 是节点角色
	Role NodeRole
	// Address 是节点地址
	Address string
	// Port 是节点端口
	Port int
	// Tags 是节点标签
	Tags map[string]string
	// Resources 是节点资源
	Resources map[resource.ResourceType]resource.ResourceQuantity
	// MaxWorkflows 是节点可运行的最大工作流数量
	MaxWorkflows int
	// MaxTasks 是节点可运行的最大任务数量
	MaxTasks int
}

// Node 表示分布式运行时节点
type Node struct {
	// config 是节点配置
	config *NodeConfig
	// status 是节点状态
	status NodeStatus
	// clusterManager 是集群管理器
	clusterManager cluster.Manager
	// resourceManager 是资源管理器
	resourceManager resource.ResourceManager
	// leaderElection 是领导者选举
	leaderElection coordination.LeaderElection
	// actorSystem 是Actor系统
	actorSystem *actor.ActorSystem
	// remoting 是远程通信
	remoting *remote.Remote
	// runningWorkflows 是正在运行的工作流
	runningWorkflows map[string]string
	// runningTasks 是正在运行的任务
	runningTasks map[string]string
	// mu 是保护状态的互斥锁
	mu sync.RWMutex
	// logger 是日志记录器
	logger *log.Logger
	// ctx 是上下文
	ctx context.Context
	// cancel 是取消函数
	cancel context.CancelFunc
}

// NewNode 创建一个新的分布式运行时节点
func NewNode(config *NodeConfig, clusterManager cluster.Manager) (*Node, error) {
	// 验证配置
	if config.ID == "" {
		config.ID = uuid.New().String()
	}
	if config.Name == "" {
		config.Name = fmt.Sprintf("node-%s", config.ID[:8])
	}
	if config.Role == "" {
		config.Role = NodeRoleMixed
	}
	if config.Address == "" {
		config.Address = "localhost"
	}
	if config.Port == 0 {
		config.Port = 8090
	}
	if config.Tags == nil {
		config.Tags = make(map[string]string)
	}
	if config.Resources == nil {
		config.Resources = make(map[resource.ResourceType]resource.ResourceQuantity)
	}
	if config.MaxWorkflows == 0 {
		config.MaxWorkflows = 10
	}
	if config.MaxTasks == 0 {
		config.MaxTasks = 100
	}

	// 创建上下文
	ctx, cancel := context.WithCancel(context.Background())

	// 创建资源管理器
	resourceManager := resource.NewClusterResourceManager(clusterManager)

	// 创建领导者选举
	leaderElection := coordination.NewClusterLeaderElection(clusterManager, "runtime-leader")

	// 创建Actor系统
	actorSystem := actor.NewActorSystem()

	// 创建节点
	node := &Node{
		config:           config,
		status:           NodeStatusInitializing,
		clusterManager:   clusterManager,
		resourceManager:  resourceManager,
		leaderElection:   leaderElection,
		actorSystem:      actorSystem,
		runningWorkflows: make(map[string]string),
		runningTasks:     make(map[string]string),
		logger:           log.New(os.Stderr, fmt.Sprintf("[Node-%s] ", config.ID[:8]), log.LstdFlags),
		ctx:              ctx,
		cancel:           cancel,
	}

	return node, nil
}

// Start 启动节点
func (n *Node) Start(ctx context.Context) error {
	n.logger.Printf("Starting node %s (%s) with role %s", n.config.Name, n.config.ID, n.config.Role)

	// 启动资源管理器
	err := n.resourceManager.Start(ctx)
	if err != nil {
		return fmt.Errorf("failed to start resource manager: %w", err)
	}

	// 注册资源
	for resourceType, quantity := range n.config.Resources {
		err := n.resourceManager.RegisterResource(ctx, resourceType, quantity)
		if err != nil {
			n.logger.Printf("Failed to register resource %s: %v", resourceType, err)
		}
	}

	// 启动领导者选举
	err = n.leaderElection.Start(ctx)
	if err != nil {
		return fmt.Errorf("failed to start leader election: %w", err)
	}

	// 设置领导权变更回调
	n.leaderElection.OnLeadershipChange(func(isLeader bool) {
		if isLeader {
			n.logger.Println("Became cluster leader")
			n.handleLeadershipAcquired(ctx)
		} else {
			n.logger.Println("Lost cluster leadership")
			n.handleLeadershipLost(ctx)
		}
	})

	// 配置远程通信
	remoteConfig := remote.Configure(n.config.Address, n.config.Port)
	n.remoting = remote.NewRemote(n.actorSystem, remoteConfig)
	err = n.remoting.Start()
	if err != nil {
		return fmt.Errorf("failed to start remote communication: %w", err)
	}

	// 更新节点状态
	n.mu.Lock()
	n.status = NodeStatusReady
	n.mu.Unlock()

	// 更新节点元数据
	err = n.updateNodeMetadata(ctx)
	if err != nil {
		n.logger.Printf("Failed to update node metadata: %v", err)
	}

	// 启动定期任务
	go n.periodicTasks()

	n.logger.Printf("Node %s started successfully", n.config.Name)
	return nil
}

// Stop 停止节点
func (n *Node) Stop(ctx context.Context) error {
	n.logger.Printf("Stopping node %s", n.config.Name)

	// 更新节点状态
	n.mu.Lock()
	n.status = NodeStatusDraining
	n.mu.Unlock()

	// 更新节点元数据
	err := n.updateNodeMetadata(ctx)
	if err != nil {
		n.logger.Printf("Failed to update node metadata: %v", err)
	}

	// 停止所有工作流
	err = n.stopAllWorkflows(ctx)
	if err != nil {
		n.logger.Printf("Failed to stop all workflows: %v", err)
	}

	// 停止远程通信
	err = n.remoting.Shutdown(true)
	if err != nil {
		n.logger.Printf("Failed to stop remote communication: %v", err)
	}

	// 停止领导者选举
	err = n.leaderElection.Stop(ctx)
	if err != nil {
		n.logger.Printf("Failed to stop leader election: %v", err)
	}

	// 停止资源管理器
	err = n.resourceManager.Stop(ctx)
	if err != nil {
		n.logger.Printf("Failed to stop resource manager: %v", err)
	}

	// 取消上下文
	n.cancel()

	n.logger.Printf("Node %s stopped successfully", n.config.Name)
	return nil
}

// GetStatus 获取节点状态
func (n *Node) GetStatus() NodeStatus {
	n.mu.RLock()
	defer n.mu.RUnlock()
	return n.status
}

// SetStatus 设置节点状态
func (n *Node) SetStatus(status NodeStatus) error {
	n.mu.Lock()
	n.status = status
	n.mu.Unlock()

	// 更新节点元数据
	return n.updateNodeMetadata(n.ctx)
}

// IsLeader 检查节点是否为领导者
func (n *Node) IsLeader() bool {
	return n.leaderElection.IsLeader()
}

// GetRunningWorkflows 获取正在运行的工作流
func (n *Node) GetRunningWorkflows() map[string]string {
	n.mu.RLock()
	defer n.mu.RUnlock()
	
	// 创建副本
	workflows := make(map[string]string)
	for id, runID := range n.runningWorkflows {
		workflows[id] = runID
	}
	
	return workflows
}

// GetRunningTasks 获取正在运行的任务
func (n *Node) GetRunningTasks() map[string]string {
	n.mu.RLock()
	defer n.mu.RUnlock()
	
	// 创建副本
	tasks := make(map[string]string)
	for id, status := range n.runningTasks {
		tasks[id] = status
	}
	
	return tasks
}

// DeployWorkflow 部署工作流
func (n *Node) DeployWorkflow(ctx context.Context, workflowID string, definition interface{}) error {
	n.mu.Lock()
	defer n.mu.Unlock()

	// 检查节点状态
	if n.status != NodeStatusReady && n.status != NodeStatusBusy {
		return fmt.Errorf("node is not ready to deploy workflows: %s", n.status)
	}

	// 检查工作流数量限制
	if len(n.runningWorkflows) >= n.config.MaxWorkflows {
		return fmt.Errorf("node has reached maximum workflow limit: %d", n.config.MaxWorkflows)
	}

	// 序列化工作流定义
	definitionBytes, err := json.Marshal(definition)
	if err != nil {
		return fmt.Errorf("failed to marshal workflow definition: %w", err)
	}

	// 创建部署消息
	deployMsg := &DeployWorkflowMessage{
		WorkflowID:  workflowID,
		Definition:  definitionBytes,
		DeployerID:  n.config.ID,
		DeployedAt:  time.Now(),
	}

	// 发送消息到工作流管理器
	future := n.actorSystem.Root.RequestFuture(n.getWorkflowManagerPID(), deployMsg, 30*time.Second)
	result, err := future.Result()
	if err != nil {
		return fmt.Errorf("failed to deploy workflow: %w", err)
	}

	// 处理响应
	response, ok := result.(*DeployWorkflowResponse)
	if !ok {
		return fmt.Errorf("unexpected response type: %T", result)
	}

	if !response.Success {
		return fmt.Errorf("failed to deploy workflow: %s", response.Error)
	}

	n.logger.Printf("Workflow %s deployed successfully", workflowID)
	return nil
}

// StartWorkflow 启动工作流
func (n *Node) StartWorkflow(ctx context.Context, workflowID string, parameters map[string]interface{}) (string, error) {
	n.mu.Lock()
	defer n.mu.Unlock()

	// 检查节点状态
	if n.status != NodeStatusReady && n.status != NodeStatusBusy {
		return "", fmt.Errorf("node is not ready to start workflows: %s", n.status)
	}

	// 序列化参数
	parametersBytes, err := json.Marshal(parameters)
	if err != nil {
		return "", fmt.Errorf("failed to marshal workflow parameters: %w", err)
	}

	// 创建启动消息
	startMsg := &StartWorkflowMessage{
		WorkflowID:  workflowID,
		Parameters:  parametersBytes,
		StarterID:   n.config.ID,
		StartedAt:   time.Now(),
	}

	// 发送消息到工作流管理器
	future := n.actorSystem.Root.RequestFuture(n.getWorkflowManagerPID(), startMsg, 30*time.Second)
	result, err := future.Result()
	if err != nil {
		return "", fmt.Errorf("failed to start workflow: %w", err)
	}

	// 处理响应
	response, ok := result.(*StartWorkflowResponse)
	if !ok {
		return "", fmt.Errorf("unexpected response type: %T", result)
	}

	if !response.Success {
		return "", fmt.Errorf("failed to start workflow: %s", response.Error)
	}

	// 更新运行中的工作流
	n.runningWorkflows[workflowID] = response.RunID

	// 更新节点状态
	if len(n.runningWorkflows) > 0 {
		n.status = NodeStatusBusy
	}

	n.logger.Printf("Workflow %s started with run ID %s", workflowID, response.RunID)
	return response.RunID, nil
}

// StopWorkflow 停止工作流
func (n *Node) StopWorkflow(ctx context.Context, workflowID string, runID string) error {
	n.mu.Lock()
	defer n.mu.Unlock()

	// 检查工作流是否在运行
	currentRunID, ok := n.runningWorkflows[workflowID]
	if !ok {
		return fmt.Errorf("workflow %s is not running on this node", workflowID)
	}

	// 检查运行ID是否匹配
	if runID != "" && runID != currentRunID {
		return fmt.Errorf("run ID mismatch: expected %s, got %s", currentRunID, runID)
	}

	// 创建停止消息
	stopMsg := &StopWorkflowMessage{
		WorkflowID: workflowID,
		RunID:      currentRunID,
		StopperID:  n.config.ID,
		StoppedAt:  time.Now(),
	}

	// 发送消息到工作流管理器
	future := n.actorSystem.Root.RequestFuture(n.getWorkflowManagerPID(), stopMsg, 30*time.Second)
	result, err := future.Result()
	if err != nil {
		return fmt.Errorf("failed to stop workflow: %w", err)
	}

	// 处理响应
	response, ok := result.(*StopWorkflowResponse)
	if !ok {
		return fmt.Errorf("unexpected response type: %T", result)
	}

	if !response.Success {
		return fmt.Errorf("failed to stop workflow: %s", response.Error)
	}

	// 更新运行中的工作流
	delete(n.runningWorkflows, workflowID)

	// 更新节点状态
	if len(n.runningWorkflows) == 0 {
		n.status = NodeStatusReady
	}

	n.logger.Printf("Workflow %s stopped successfully", workflowID)
	return nil
}

// GetWorkflowStatus 获取工作流状态
func (n *Node) GetWorkflowStatus(ctx context.Context, workflowID string, runID string) (interface{}, error) {
	// 检查工作流是否在运行
	n.mu.RLock()
	currentRunID, ok := n.runningWorkflows[workflowID]
	n.mu.RUnlock()

	if !ok {
		return nil, fmt.Errorf("workflow %s is not running on this node", workflowID)
	}

	// 检查运行ID是否匹配
	if runID != "" && runID != currentRunID {
		return nil, fmt.Errorf("run ID mismatch: expected %s, got %s", currentRunID, runID)
	}

	// 创建状态查询消息
	statusMsg := &GetWorkflowStatusMessage{
		WorkflowID: workflowID,
		RunID:      currentRunID,
	}

	// 发送消息到工作流管理器
	future := n.actorSystem.Root.RequestFuture(n.getWorkflowManagerPID(), statusMsg, 30*time.Second)
	result, err := future.Result()
	if err != nil {
		return nil, fmt.Errorf("failed to get workflow status: %w", err)
	}

	// 处理响应
	response, ok := result.(*GetWorkflowStatusResponse)
	if !ok {
		return nil, fmt.Errorf("unexpected response type: %T", result)
	}

	if !response.Success {
		return nil, fmt.Errorf("failed to get workflow status: %s", response.Error)
	}

	return response.Status, nil
}

// handleLeadershipAcquired 处理获得领导权
func (n *Node) handleLeadershipAcquired(ctx context.Context) {
	n.logger.Println("Handling leadership acquisition")

	// 更新节点元数据
	err := n.updateNodeMetadata(ctx)
	if err != nil {
		n.logger.Printf("Failed to update node metadata: %v", err)
	}

	// 启动集群管理Actor
	props := actor.PropsFromProducer(func() actor.Actor {
		return NewClusterManagerActor(n)
	})

	pid, err := n.actorSystem.Root.SpawnNamed(props, "cluster-manager")
	if err != nil {
		n.logger.Printf("Failed to spawn cluster manager actor: %v", err)
	} else {
		n.logger.Printf("Cluster manager actor started with PID: %s", pid.String())
	}
}

// handleLeadershipLost 处理失去领导权
func (n *Node) handleLeadershipLost(ctx context.Context) {
	n.logger.Println("Handling leadership loss")

	// 更新节点元数据
	err := n.updateNodeMetadata(ctx)
	if err != nil {
		n.logger.Printf("Failed to update node metadata: %v", err)
	}

	// 停止集群管理Actor
	pid := n.actorSystem.Root.NewPID("nonhost", "cluster-manager")
	n.actorSystem.Root.Stop(pid)
}

// updateNodeMetadata 更新节点元数据
func (n *Node) updateNodeMetadata(ctx context.Context) error {
	n.mu.RLock()
	defer n.mu.RUnlock()

	// 创建节点元数据
	metadata := map[string]string{
		"status":           string(n.status),
		"role":             string(n.config.Role),
		"workflows_count":  fmt.Sprintf("%d", len(n.runningWorkflows)),
		"tasks_count":      fmt.Sprintf("%d", len(n.runningTasks)),
		"max_workflows":    fmt.Sprintf("%d", n.config.MaxWorkflows),
		"max_tasks":        fmt.Sprintf("%d", n.config.MaxTasks),
		"is_leader":        fmt.Sprintf("%t", n.leaderElection.IsLeader()),
		"version":          "1.0.0",
		"last_updated":     time.Now().Format(time.RFC3339),
	}

	// 添加自定义标签
	for k, v := range n.config.Tags {
		metadata[k] = v
	}

	// 更新节点元数据
	return n.clusterManager.UpdateMetadata(ctx, metadata)
}

// periodicTasks 执行定期任务
func (n *Node) periodicTasks() {
	// 创建定时器
	metadataTicker := time.NewTicker(30 * time.Second)
	healthTicker := time.NewTicker(60 * time.Second)
	defer metadataTicker.Stop()
	defer healthTicker.Stop()

	for {
		select {
		case <-n.ctx.Done():
			return
		case <-metadataTicker.C:
			// 更新节点元数据
			err := n.updateNodeMetadata(n.ctx)
			if err != nil {
				n.logger.Printf("Failed to update node metadata: %v", err)
			}
		case <-healthTicker.C:
			// 执行健康检查
			n.performHealthCheck()
		}
	}
}

// performHealthCheck 执行健康检查
func (n *Node) performHealthCheck() {
	n.logger.Println("Performing health check")

	// 检查资源使用情况
	allocatedResources, err := n.resourceManager.GetAllocatedResources(n.ctx)
	if err != nil {
		n.logger.Printf("Failed to get allocated resources: %v", err)
	} else {
		for resourceType, allocations := range allocatedResources {
			n.logger.Printf("Resource %s allocations: %v", resourceType, allocations)
		}
	}

	// 检查工作流状态
	n.mu.RLock()
	workflowCount := len(n.runningWorkflows)
	taskCount := len(n.runningTasks)
	n.mu.RUnlock()

	n.logger.Printf("Running workflows: %d, Running tasks: %d", workflowCount, taskCount)

	// 检查集群状态
	nodes, err := n.clusterManager.GetNodes(n.ctx)
	if err != nil {
		n.logger.Printf("Failed to get cluster nodes: %v", err)
	} else {
		n.logger.Printf("Cluster nodes: %d", len(nodes))
	}

	// 检查领导者状态
	isLeader := n.leaderElection.IsLeader()
	n.logger.Printf("Is leader: %t", isLeader)
}

// stopAllWorkflows 停止所有工作流
func (n *Node) stopAllWorkflows(ctx context.Context) error {
	n.mu.Lock()
	workflows := make(map[string]string)
	for id, runID := range n.runningWorkflows {
		workflows[id] = runID
	}
	n.mu.Unlock()

	// 停止所有工作流
	for id, runID := range workflows {
		err := n.StopWorkflow(ctx, id, runID)
		if err != nil {
			n.logger.Printf("Failed to stop workflow %s: %v", id, err)
		}
	}

	return nil
}

// getWorkflowManagerPID 获取工作流管理器PID
func (n *Node) getWorkflowManagerPID() *actor.PID {
	return n.actorSystem.Root.NewPID("nonhost", "workflow-manager")
}
